package com.bigdata.spark.core.acc

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

object Bc01 {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local").setAppName("Bc")
    val sc = new SparkContext(sparkConf)


    // 闭包数据，都是一Task为单位发送的，每个任务中包含闭包数据
    // 这样可能会导致，一个Executor中含有大量重复的数据，并且占用大量的内存
    // Executor其实就是一个JVM，所以在启动时，会自动分配内存
    // 完全可以将任务中的闭包数据放置在Executor的内存中，达到共享的目的
    // Spark中的广播变量就可以将闭包的数据保存到Executor的内存中
    // Spark中的广播变量不能更改
    // TODO - 广播变量：分布式共享只读变量

    val rdd1 = sc.makeRDD(List(
      ("a", 1), ("b", 2), ("c", 3)
    ))

    //    val rdd2 = sc.makeRDD(List(
    //      ("a",4),("b",5),("c",6)
    //    ))

    val map = mutable.Map(("a", 4), ("b", 5), ("c", 6))

    // join会导致数据量几何增长，并且会影响shuffle的性能，不推荐使用
    //val joinRDD: RDD[(String, (Int, Int))] = rdd1.join(rdd2)
    //joinRDD.collect().foreach(println)
    // ("a",1),("b",2),("c",3)
    // (a,(1,4)),(b,(2,5)),(c,(3,6))
    rdd1.map {
      case (w, c) => {
        val l = map.getOrElse(w, 0)
        (w, (c, l))
      }
    }.collect().foreach(println)


    sc.stop()
  }
}
